package com.example.demo.kafka;

import com.example.demo.exception.BaseServiceException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Arrays;
import java.util.List;

/**
 * ******************************************************
 *
 * @author liugh9
 * @version 1.0
 * @classname KafkaController
 * @description
 * @date 2022/12/01 14:54
 * <p>
 * ******************************************************
 */
@RestController
@RequestMapping("/kafka/produce/")
public class KafkaController {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaController.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /***
     * 发送消息
     * topic:要发送的队列
     * msg:发送的消息
     */
    @GetMapping(value = "/send")
    public String send() {
        //消息发送
        List<String> strings = Arrays.asList("aaa", "bbb", "ccc");
        List<String> ints = Arrays.asList("111", "222", "333");
        for (String itemString : strings) {
            for (String intItem : ints) {
                sendMessage(itemString + intItem);
            }
        }
        return "SUCCESS";
    }


    public void sendMessage(String message) {
        String topicName = "acc-alarm";
        ObjectMapper objectMapper = new ObjectMapper();
//        String message = "";
//        try {
//            message = objectMapper.writeValueAsString(kafkaMessageVo);
//        } catch (JsonProcessingException e) {
//            logger.error("entity change to json fail...");
//            throw new BaseServiceException("entity change to json fail...");
//        }

        // topic key名称 data消息数据
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topicName, message);

        // 开启消息监听
        //发送成功后回调
        String finalMessage = message;
        SuccessCallback<SendResult<String, String>> successCallback = new SuccessCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                LOGGER.info("发送消息成功 - " + " value:" + finalMessage);
            }
        };
        //发送失败回调
        FailureCallback failureCallback = new FailureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                // TODO 后续跟进发送失败方案
                LOGGER.error("发送消息失败-添加消息到发送失败日志中 - " + " value:" + finalMessage);
            }
        };

        listenableFuture.addCallback(successCallback, failureCallback);
    }
}
